// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: change E2E packet format
// feature: develop dispatcher
// feature: Split history as an independent common module
// feature: discovery support client and server
// feature: support lookup writer/reader proxy info with cmd-tool
// feature: add statictic lookup
// feature: E2E New Requirement Development
// ------------------------------------------------------------------

#ifndef INCLUDE_EDDS_RTPS_READER_STATELESSREADER_H_
#define INCLUDE_EDDS_RTPS_READER_STATELESSREADER_H_

#ifndef DOXYGEN_SHOULD_SKIP_THIS_PUBLIC

#include <edds/rtps/reader/RTPSReader.h>

#include <mutex>
#include <map>
#include <unordered_map>

#include <ertps/utils/collections/ResourceLimitedVector.hpp>

namespace evbs {
namespace ertps {
namespace rtps {

/**
 * Class StatelessReader, specialization of the RTPSReader for Best Effort Readers.
 * @ingroup READER_MODULE
 */
class StatelessReader : public RTPSReader {
    friend class RTPSParticipantImpl;

 public:
    virtual ~StatelessReader();

 protected:
    StatelessReader(RTPSParticipantImpl* pimpl, const GUID_t& guid, const ReaderAttributes& att, ReaderHistory* hist,
                    vbs::common::ReaderListener* listen = nullptr);

    StatelessReader(RTPSParticipantImpl* pimpl, const GUID_t& guid, const ReaderAttributes& att,
                    const std::shared_ptr<vbs::common::IChangePool>& change_pool, ReaderHistory* hist,
                    vbs::common::ReaderListener* listen = nullptr);

    StatelessReader(RTPSParticipantImpl* pimpl, const GUID_t& guid, const ReaderAttributes& att,
                    const std::shared_ptr<vbs::common::IPayloadPool>& payload_pool, ReaderHistory* hist,
                    vbs::common::ReaderListener* listen = nullptr);

    StatelessReader(RTPSParticipantImpl* pimpl, const GUID_t& guid, const ReaderAttributes& att,
                    const std::shared_ptr<vbs::common::IPayloadPool>& payload_pool,
                    const std::shared_ptr<vbs::common::IChangePool>& change_pool, ReaderHistory* hist,
                    vbs::common::ReaderListener* listen = nullptr);

 public:
    /**
     * Add a matched writer represented by a WriterProxyData object.
     * @param wdata Pointer to the WPD object to add.
     * @return True if correctly added.
     */
    bool matched_writer_add(WriterProxyData& wdata) override;

    /**
     * Remove a WriterProxyData from the matached writers.
     * @param writer_guid GUID of the writer to remove.
     * @param removed_by_lease true it the writer was removed due to lease duration.
     * @return True if correct.
     */
    bool matched_writer_remove(const GUID_t& writer_guid, bool removed_by_lease = false) override;

    /**
     * Tells us if a specific Writer is matched against this reader.
     * @param writer_guid GUID of the writer to check.
     * @return True if it is matched.
     */
    bool matched_writer_is_matched(const GUID_t& writer_guid) override;

    virtual bool processDataBatchMsg(CacheChange_t* change);

    /**
     * Processes a new DATA message.
     *
     * @param change Pointer to the CacheChange_t.
     * @return true if the reader accepts messages from the.
     */
    bool processDataMsg(CacheChange_t* change) override;

    /**
     * Processes a new DATA FRAG message.
     *
     * @param change Pointer to the CacheChange_t.
     * @param sampleSize Size of the complete, assembled message.
     * @param fragmentStartingNum Starting number of this particular message.
     * @param fragmentsInSubmessage Number of fragments on this particular message.
     * @return true if the reader accepts message.
     */
    bool processDataFragMsg(CacheChange_t* change, uint32_t sampleSize, uint32_t fragmentStartingNum,
                            uint16_t fragmentsInSubmessage) override;

    /**
     * Processes a new HEARTBEAT message.
     *
     * @return true if the reader accepts messages from the.
     */
    bool processHeartbeatMsg(const GUID_t& writerGUID, uint32_t hbCount, const SequenceNumber_t& firstSN,
                             const SequenceNumber_t& lastSN, bool finalFlag, bool livelinessFlag) override;

    bool processGapMsg(const GUID_t& writerGUID, const SequenceNumber_t& gapStart,
                       const SequenceNumberSet_t& gapList) override;

    bool processE2EMsg(const E2EProfile04_t& e2eHeader) override;

    /**
     * This method is called when a new change is received. This method calls the received_change of
     * the History and depending on the implementation performs different actions.
     * @param a_change Pointer of the change to add.
     * @return True if added.
     */
    inline bool change_received(CacheChange_t* a_change);

    /**
     * Take the next CacheChange_t from the history;
     * @param change Pointer to pointer of CacheChange_t
     * @param wpout Pointer to pointer of the matched writer proxy
     * @return True if read.
     */
    bool nextUntakenCache(CacheChange_t** change, WriterProxy** wpout = nullptr) override;

    /**
     * Get the number of matched writers
     * @return Number of matched writers
     */
    size_t getMatchedWritersSize() const override { return matched_writers_.size(); }

    /*!
     * @brief Returns there is a clean state with all Writers.
     * StatelessReader allways return true;
     * @return true
     */
    bool is_in_clean_state() override { return true; }
    /**
     * Get the RTPS participant
     * @return Associated RTPS participant
     */
    inline RTPSParticipantImpl* getRTPSParticipant() const { return mp_RTPSParticipant; }

    /**
     * @brief Assert liveliness of remote writer
     * @param guid The guid of the remote writer
     */
    void assert_writer_liveliness(const GUID_t& writer) const override;

    /**
     * Called just before a change is going to be deserialized.
     * @param [in]  change            Pointer to the change being accessed.
     * @param [out] wp                Writer proxy the @c change belongs to.
     * @param [out] is_future_change  Whether the change is in the future (i.e. there are
     *                                earlier unreceived changes from the same writer).
     *
     * @return Whether the change is still valid or not.
     */
    bool begin_sample_access_nts(CacheChange_t* change, /*WriterProxy*& wp,*/ bool& is_future_change) override;

    /**
     * Called after the change has been deserialized.
     * @param [in] change        Pointer to the change being accessed.
     * @param [in] wp            Writer proxy the @c change belongs to.
     * @param [in] mark_as_read  Whether the @c change should be marked as read or not.
     */
    void end_sample_access_nts(CacheChange_t* change, /*WriterProxy*& wp,*/ bool mark_as_read) override;

    /**
     * Called when the user has retrieved a change from the history.
     * @param change Pointer to the change to ACK
     * @param writer Writer proxy of the \c change.
     * @param mark_as_read Whether the \c change should be marked as read or not
     */
    void change_read_by_user(CacheChange_t* const change, const WriterProxy* const writer,
                             const bool mark_as_read = true) override;

    edds::dds::builtin::StatisticMatchGuids get_remote_guids() override;

    edds::dds::builtin::StatisticProxyInfos get_proxy_infos() override;

 private:
    struct RemoteWriterInfo_t {
        GUID_t guid;
        GUID_t persistence_guid;
        bool has_manual_topic_liveliness = false;
        CacheChange_t* fragmented_change = nullptr;
        uint32_t ownership_strength;
        RemoteLocatorList remote_locator_list;
        Time_t start_time;
    };

    inline bool acceptMsgFrom(const GUID_t& entityId, ChangeKind_t change_kind);

    inline bool thereIsUpperRecordOf(const GUID_t& guid, const SequenceNumber_t& seq);

    /**
     * @brief A method to check if a matched writer has manual_by_topic liveliness
     * @param guid The guid of the remote writer
     * @return True if writer has manual_by_topic livelinesss
     */
    bool writer_has_manual_liveliness(const GUID_t& guid);

    inline void report_sample_lost(SampleLostStatusKind statusKind, uint32_t lost_num = 1U);

    //! List of GUID_t os matched writers.
    //! Is only used in the Discovery, to correctly notify the user using
    //! SubscriptionListener::onSubscriptionMatched();
    ResourceLimitedVector<RemoteWriterInfo_t> matched_writers_;
    //  uint32_t AppID;
    std::unordered_map<std::string, SequenceNumber_t> counter_map_;
    std::unordered_map<std::string, bool> isNotFirstReceive_map_;
};

} /* namespace rtps */
} /* namespace ertps */
} /* namespace evbs */

#endif  // ifndef DOXYGEN_SHOULD_SKIP_THIS_PUBLIC

#endif  // INCLUDE_EDDS_RTPS_READER_STATELESSREADER_H_
